/*
 * Copyright 2009 Mike Cumings
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.kenai.jbosh;

import com.kenai.jbosh.ComposableBody.Builder;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * BOSH Client session instance. Each communication session with a remote
 * connection manager is represented and handled by an instance of this class.
 * This is the main entry point for client-side communications. To create a new
 * session, a client configuration must first be created and then used to create
 * a client instance:
 * 
 * <pre>
 * BOSHClientConfig cfg = BOSHClientConfig.Builder
 * 		.create(&quot;http://server:1234/httpbind&quot;, &quot;jabber.org&quot;)
 * 		.setFrom(&quot;user@jabber.org&quot;).build();
 * BOSHClient client = BOSHClient.create(cfg);
 * </pre>
 * 
 * Additional client configuration options are available. See the
 * {@code BOSHClientConfig.Builder} class for more information.
 * <p/>
 * Once a {@code BOSHClient} instance has been created, communication with the
 * remote connection manager can begin. No attempt will be made to establish a
 * connection to the connection manager until the first call is made to the
 * {@code send(ComposableBody)} method. Note that it is possible to send an
 * empty body to cause an immediate connection attempt to the connection
 * manager. Sending an empty message would look like the following:
 * 
 * <pre>
 * client.send(ComposableBody.builder().build());
 * </pre>
 * 
 * For more information on creating body messages with content, see the
 * {@code ComposableBody.Builder} class documentation.
 * <p/>
 * Once a session has been successfully started, the client instance can be used
 * to send arbitrary payload data. All aspects of the BOSH protocol involving
 * setting and processing attributes in the BOSH namespace will be handled by
 * the client code transparently and behind the scenes. The user of the client
 * instance can therefore concentrate entirely on the content of the message
 * payload, leaving the semantics of the BOSH protocol to the client
 * implementation.
 * <p/>
 * To be notified of incoming messages from the remote connection manager, a
 * {@code BOSHClientResponseListener} should be added to the client instance.
 * All incoming messages will be published to all response listeners as they
 * arrive and are processed. As with the transmission of payload data via the
 * {@code send(ComposableBody)} method, there is no need to worry about handling
 * of the BOSH attributes, since this is handled behind the scenes.
 * <p/>
 * If the connection to the remote connection manager is terminated (either
 * explicitly or due to a terminal condition of some sort), all connection
 * listeners will be notified. After the connection has been closed, the client
 * instance is considered dead and a new one must be created in order to resume
 * communications with the remote server.
 * <p/>
 * Instances of this class are thread-safe.
 * 
 * @see BOSHClientConfig.Builder
 * @see BOSHClientResponseListener
 * @see BOSHClientConnListener
 * @see ComposableBody.Builder
 */
public final class BOSHClient {

	/**
	 * Logger.
	 */
	private static final Logger LOG = Logger.getLogger(BOSHClient.class
			.getName());

	/**
	 * Value of the 'type' attribute used for session termination.
	 */
	private static final String TERMINATE = "terminate";

	/**
	 * Value of the 'type' attribute used for recoverable errors.
	 */
	private static final String ERROR = "error";

	/**
	 * Message to use for interrupted exceptions.
	 */
	private static final String INTERRUPTED = "Interrupted";

	/**
	 * Message used for unhandled exceptions.
	 */
	private static final String UNHANDLED = "Unhandled Exception";

	/**
	 * Message used whena null listener is detected.
	 */
	private static final String NULL_LISTENER = "Listener may not b enull";

	/**
	 * Default empty request delay.
	 */
	private static final int DEFAULT_EMPTY_REQUEST_DELAY = 100;

	/**
	 * Amount of time to wait before sending an empty request, in milliseconds.
	 */
	private static final int EMPTY_REQUEST_DELAY = Integer.getInteger(
			BOSHClient.class.getName() + ".emptyRequestDelay",
			DEFAULT_EMPTY_REQUEST_DELAY);

	/**
	 * Default value for the pause margin.
	 */
	private static final int DEFAULT_PAUSE_MARGIN = 500;

	/**
	 * The amount of time in milliseconds which will be reserved as a safety
	 * margin when scheduling empty requests against a maxpause value. This
	 * should give us enough time to build the message and transport it to the
	 * remote host.
	 */
	private static final int PAUSE_MARGIN = Integer.getInteger(
			BOSHClient.class.getName() + ".pauseMargin", DEFAULT_PAUSE_MARGIN);

	/**
	 * Flag indicating whether or not we want to perform assertions.
	 */
	private static final boolean ASSERTIONS;

	/**
	 * Connection listeners.
	 */
	private final Set<BOSHClientConnListener> connListeners = new CopyOnWriteArraySet<BOSHClientConnListener>();

	/**
	 * Request listeners.
	 */
	private final Set<BOSHClientRequestListener> requestListeners = new CopyOnWriteArraySet<BOSHClientRequestListener>();

	/**
	 * Response listeners.
	 */
	private final Set<BOSHClientResponseListener> responseListeners = new CopyOnWriteArraySet<BOSHClientResponseListener>();

	/**
	 * Lock instance.
	 */
	private final ReentrantLock lock = new ReentrantLock();

	/**
	 * Condition indicating that there are messages to be exchanged.
	 */
	private final Condition notEmpty = lock.newCondition();

	/**
	 * Condition indicating that there are available slots for sending messages.
	 */
	private final Condition notFull = lock.newCondition();

	/**
	 * Condition indicating that there are no outstanding connections.
	 */
	private final Condition drained = lock.newCondition();

	/**
	 * Session configuration.
	 */
	private final BOSHClientConfig cfg;

	/**
	 * Processor thread runnable instance.
	 */
	private final Runnable procRunnable = new Runnable() {
		/**
		 * Process incoming messages.
		 */
		public void run() {
			processMessages();
		}
	};

	/**
	 * Processor thread runnable instance.
	 */
	private final Runnable emptyRequestRunnable = new Runnable() {
		/**
		 * Process incoming messages.
		 */
		public void run() {
			sendEmptyRequest();
		}
	};

	/**
	 * HTTPSender instance.
	 */
	private final HTTPSender httpSender = new ApacheHTTPSender();

	/**
	 * Storage for test hook implementation.
	 */
	private final AtomicReference<ExchangeInterceptor> exchInterceptor = new AtomicReference<ExchangeInterceptor>();

	/**
	 * Request ID sequence to use for the session.
	 */
	private final RequestIDSequence requestIDSeq = new RequestIDSequence();

	/**
	 * ScheduledExcecutor to use for deferred tasks.
	 */
	private final ScheduledExecutorService schedExec = Executors
			.newSingleThreadScheduledExecutor();

	/************************************************************
	 * The following vars must be accessed via the lock instance.
	 */

	/**
	 * Thread which is used to process responses from the connection manager.
	 * Becomes null when session is terminated.
	 */
	private Thread procThread;

	/**
	 * Future for sending a deferred empty request, if needed.
	 */
	private ScheduledFuture emptyRequestFuture;

	/**
	 * Connection Manager session parameters. Only available when in a connected
	 * state.
	 */
	private CMSessionParams cmParams;

	/**
	 * List of active/outstanding requests.
	 */
	private Queue<HTTPExchange> exchanges = new LinkedList<HTTPExchange>();

	/**
	 * Set of RIDs which have been received, for the purpose of sending response
	 * acknowledgements.
	 */
	private SortedSet<Long> pendingResponseAcks = new TreeSet<Long>();

	/**
	 * The highest RID that we've already received a response for. This value is
	 * used to implement response acks.
	 */
	private Long responseAck = Long.valueOf(-1L);

	/**
	 * List of requests which have been made but not yet acknowledged. This list
	 * remains unpopulated if the CM is not acking requests.
	 */
	private List<ComposableBody> pendingRequestAcks = new ArrayList<ComposableBody>();

	// /////////////////////////////////////////////////////////////////////////
	// Classes:

	/**
	 * Class used in testing to dynamically manipulate received exchanges at
	 * test runtime.
	 */
	abstract static class ExchangeInterceptor {
		/**
		 * Limit construction.
		 */
		ExchangeInterceptor() {
			// Empty;
		}

		/**
		 * Hook to manipulate an HTTPExchange as is is about to be processed.
		 * 
		 * @param exch
		 *            original exchange that would be processed
		 * @return replacement exchange instance, or {@code null} to skip
		 *         processing of this exchange
		 */
		abstract HTTPExchange interceptExchange(final HTTPExchange exch);
	}

	// /////////////////////////////////////////////////////////////////////////
	// Constructors:

	/**
	 * Determine whether or not we should perform assertions. Assertions can be
	 * specified via system property explicitly, or defaulted to the JVM
	 * assertions status.
	 */
	static {
		final String prop = BOSHClient.class.getSimpleName()
				+ ".assertionsEnabled";
		boolean enabled = false;
		if (System.getProperty(prop) == null) {
			assert enabled = true;
		} else {
			enabled = Boolean.getBoolean(prop);
		}
		ASSERTIONS = enabled;
	}

	/**
	 * Prevent direct construction.
	 */
	private BOSHClient(final BOSHClientConfig sessCfg) {
		cfg = sessCfg;
		init();
	}

	// /////////////////////////////////////////////////////////////////////////
	// Public methods:

	/**
	 * Create a new BOSH client session using the client configuration
	 * information provided.
	 * 
	 * @param clientCfg
	 *            session configuration
	 * @return BOSH session instance
	 */
	public static BOSHClient create(final BOSHClientConfig clientCfg) {
		if (clientCfg == null) {
			throw (new IllegalArgumentException(
					"Client configuration may not be null"));
		}
		return new BOSHClient(clientCfg);
	}

	/**
	 * Get the client configuration that was used to create this client
	 * instance.
	 * 
	 * @return client configuration
	 */
	public BOSHClientConfig getBOSHClientConfig() {
		return cfg;
	}

	/**
	 * Adds a connection listener to the session.
	 * 
	 * @param listener
	 *            connection listener to add, if not already added
	 */
	public void addBOSHClientConnListener(final BOSHClientConnListener listener) {
		if (listener == null) {
			throw (new IllegalArgumentException(NULL_LISTENER));
		}
		connListeners.add(listener);
	}

	/**
	 * Removes a connection listener from the session.
	 * 
	 * @param listener
	 *            connection listener to remove, if previously added
	 */
	public void removeBOSHClientConnListener(
			final BOSHClientConnListener listener) {
		if (listener == null) {
			throw (new IllegalArgumentException(NULL_LISTENER));
		}
		connListeners.remove(listener);
	}

	/**
	 * Adds a request message listener to the session.
	 * 
	 * @param listener
	 *            request listener to add, if not already added
	 */
	public void addBOSHClientRequestListener(
			final BOSHClientRequestListener listener) {
		if (listener == null) {
			throw (new IllegalArgumentException(NULL_LISTENER));
		}
		requestListeners.add(listener);
	}

	/**
	 * Removes a request message listener from the session, if previously added.
	 * 
	 * @param listener
	 *            instance to remove
	 */
	public void removeBOSHClientRequestListener(
			final BOSHClientRequestListener listener) {
		if (listener == null) {
			throw (new IllegalArgumentException(NULL_LISTENER));
		}
		requestListeners.remove(listener);
	}

	/**
	 * Adds a response message listener to the session.
	 * 
	 * @param listener
	 *            response listener to add, if not already added
	 */
	public void addBOSHClientResponseListener(
			final BOSHClientResponseListener listener) {
		if (listener == null) {
			throw (new IllegalArgumentException(NULL_LISTENER));
		}
		responseListeners.add(listener);
	}

	/**
	 * Removes a response message listener from the session, if previously
	 * added.
	 * 
	 * @param listener
	 *            instance to remove
	 */
	public void removeBOSHClientResponseListener(
			final BOSHClientResponseListener listener) {
		if (listener == null) {
			throw (new IllegalArgumentException(NULL_LISTENER));
		}
		responseListeners.remove(listener);
	}

	/**
	 * Send the provided message data to the remote connection manager. The
	 * provided message body does not need to have any BOSH-specific attribute
	 * information set. It only needs to contain the actual message payload that
	 * should be delivered to the remote server.
	 * <p/>
	 * The first call to this method will result in a connection attempt to the
	 * remote connection manager. Subsequent calls to this method will block
	 * until the underlying session state allows for the message to be
	 * transmitted. In certain scenarios - such as when the maximum number of
	 * outbound connections has been reached - calls to this method will block
	 * for short periods of time.
	 * 
	 * @param body
	 *            message data to send to remote server
	 * @throws BOSHException
	 *             on message transmission failure
	 */
	public void send(final ComposableBody body) throws BOSHException {
		assertUnlocked();
		if (body == null) {
			throw (new IllegalArgumentException("Message body may not be null"));
		}

		HTTPExchange exch;
		CMSessionParams params;
		lock.lock();
		try {
			blockUntilSendable(body);
			if (!isWorking() && !isTermination(body)) {
				throw (new BOSHException(
						"Cannot send message when session is closed"));
			}

			long rid = requestIDSeq.getNextRID();
			ComposableBody request = body;
			params = cmParams;
			if (params == null && exchanges.isEmpty()) {
				// This is the first message being sent
				request = applySessionCreationRequest(rid, body);
			} else {
				request = applySessionData(rid, body);
				if (cmParams.isAckingRequests()) {
					pendingRequestAcks.add(request);
				}
			}
			exch = new HTTPExchange(request);
			exchanges.add(exch);
			notEmpty.signalAll();
			clearEmptyRequest();
		} finally {
			lock.unlock();
		}
		AbstractBody finalReq = exch.getRequest();
		HTTPResponse resp = httpSender.send(params, finalReq);
		exch.setHTTPResponse(resp);
		fireRequestSent(finalReq);
	}

	/**
	 * Attempt to pause the current session. When supported by the remote
	 * connection manager, pausing the session will result in the connection
	 * manager closing out all outstanding requests (including the pause
	 * request) and increases the inactivity timeout of the session. The exact
	 * value of the temporary timeout is dependent upon the connection manager.
	 * This method should be used if a client encounters an exceptional
	 * temporary situation during which it will be unable to send requests to
	 * the connection manager for a period of time greater than the maximum
	 * inactivity period.
	 * 
	 * The session will revert back to it's normal, unpaused state when the
	 * client sends it's next message.
	 * 
	 * @return {@code true} if the connection manager supports session pausing,
	 *         {@code false} if the connection manager does not support session
	 *         pausing or if the session has not yet been established
	 */
	public boolean pause() {
		assertUnlocked();
		lock.lock();
		AttrMaxPause maxPause = null;
		try {
			if (cmParams == null) {
				return false;
			}

			maxPause = cmParams.getMaxPause();
			if (maxPause == null) {
				return false;
			}
		} finally {
			lock.unlock();
		}
		try {
			send(ComposableBody.builder()
					.setAttribute(Attributes.PAUSE, maxPause.toString())
					.build());
		} catch (BOSHException boshx) {
			LOG.log(Level.FINEST, "Could not send pause", boshx);
		}
		return true;
	}

	/**
	 * End the BOSH session by disconnecting from the remote BOSH connection
	 * manager.
	 * 
	 * @throws BOSHException
	 *             when termination message cannot be sent
	 */
	public void disconnect() throws BOSHException {
		disconnect(ComposableBody.builder().build());
	}

	/**
	 * End the BOSH session by disconnecting from the remote BOSH connection
	 * manager, sending the provided content in the final connection termination
	 * message.
	 * 
	 * @param msg
	 *            final message to send
	 * @throws BOSHException
	 *             when termination message cannot be sent
	 */
	public void disconnect(final ComposableBody msg) throws BOSHException {
		if (msg == null) {
			throw (new IllegalArgumentException("Message body may not be null"));
		}

		Builder builder = msg.rebuild();
		builder.setAttribute(Attributes.TYPE, TERMINATE);
		send(builder.build());
	}

	/**
	 * Forcibly close this client session instance. The preferred mechanism to
	 * close the connection is to send a disconnect message and wait for organic
	 * termination. Calling this method simply shuts down the local session
	 * without sending a termination message, releasing all resources associated
	 * with the session.
	 */
	public void close() {
		dispose(new BOSHException("Session explicitly closed by caller"));
	}

	// /////////////////////////////////////////////////////////////////////////
	// Package-private methods:

	/**
	 * Get the current CM session params.
	 * 
	 * @return current session params, or {@code null}
	 */
	CMSessionParams getCMSessionParams() {
		lock.lock();
		try {
			return cmParams;
		} finally {
			lock.unlock();
		}
	}

	/**
	 * Wait until no more messages are waiting to be processed.
	 */
	void drain() {
		lock.lock();
		try {
			LOG.finest("Waiting while draining...");
			while (isWorking()
					&& (emptyRequestFuture == null || emptyRequestFuture
							.isDone())) {
				try {
					drained.await();
				} catch (InterruptedException intx) {
					LOG.log(Level.FINEST, INTERRUPTED, intx);
				}
			}
			LOG.finest("Drained");
		} finally {
			lock.unlock();
		}
	}

	/**
	 * Test method used to forcibly discard next exchange.
	 * 
	 * @param interceptor
	 *            exchange interceptor
	 */
	void setExchangeInterceptor(final ExchangeInterceptor interceptor) {
		exchInterceptor.set(interceptor);
	}

	// /////////////////////////////////////////////////////////////////////////
	// Private methods:

	/**
	 * Initialize the session. This initializes the underlying HTTP transport
	 * implementation and starts the receive thread.
	 */
	private void init() {
		assertUnlocked();

		lock.lock();
		try {
			httpSender.init(cfg);
			procThread = new Thread(procRunnable);
			procThread.setDaemon(true);
			procThread.setName(BOSHClient.class.getSimpleName() + "["
					+ System.identityHashCode(this) + "]: Receive thread");
			procThread.start();
		} finally {
			lock.unlock();
		}
	}

	/**
	 * Destroy this session.
	 * 
	 * @param cause
	 *            the reason for the session termination, or {@code null} for
	 *            normal termination
	 */
	private void dispose(final Throwable cause) {
		assertUnlocked();

		lock.lock();
		try {
			if (procThread == null) {
				// Already disposed
				return;
			}
			procThread = null;
		} finally {
			lock.unlock();
		}

		if (cause == null) {
			fireConnectionClosed();
		} else {
			fireConnectionClosedOnError(cause);
		}

		lock.lock();
		try {
			clearEmptyRequest();
			exchanges = null;
			cmParams = null;
			pendingResponseAcks = null;
			pendingRequestAcks = null;
			notEmpty.signalAll();
			notFull.signalAll();
			drained.signalAll();
		} finally {
			lock.unlock();
		}

		httpSender.destroy();
		schedExec.shutdownNow();
	}

	/**
	 * Determines if the message body specified indicates a request to pause the
	 * session.
	 * 
	 * @param msg
	 *            message to evaluate
	 * @return {@code true} if the message is a pause request, {@code false}
	 *         otherwise
	 */
	private static boolean isPause(final AbstractBody msg) {
		return msg.getAttribute(Attributes.PAUSE) != null;
	}

	/**
	 * Determines if the message body specified indicates a termination of the
	 * session.
	 * 
	 * @param msg
	 *            message to evaluate
	 * @return {@code true} if the message is a session termination,
	 *         {@code false} otherwise
	 */
	private static boolean isTermination(final AbstractBody msg) {
		return TERMINATE.equals(msg.getAttribute(Attributes.TYPE));
	}

	/**
	 * Evaluates the HTTP response code and response message and returns the
	 * terminal binding condition that it describes, if any.
	 * 
	 * @param respCode
	 *            HTTP response code
	 * @param respBody
	 *            response body
	 * @return terminal binding condition, or {@code null} if not a terminal
	 *         binding condition message
	 */
	private TerminalBindingCondition getTerminalBindingCondition(
			final int respCode, final AbstractBody respBody) {
		assertLocked();

		if (isTermination(respBody)) {
			String str = respBody.getAttribute(Attributes.CONDITION);
			return TerminalBindingCondition.forString(str);
		}
		// Check for deprecated HTTP Error Conditions
		if (cmParams != null && cmParams.getVersion() == null) {
			return TerminalBindingCondition.forHTTPResponseCode(respCode);
		}
		return null;
	}

	/**
	 * Determines if the message specified is immediately sendable or if it
	 * needs to block until the session state changes.
	 * 
	 * @param msg
	 *            message to evaluate
	 * @return {@code true} if the message can be immediately sent,
	 *         {@code false} otherwise
	 */
	private boolean isImmediatelySendable(final AbstractBody msg) {
		assertLocked();

		if (cmParams == null) {
			// block if we're waiting for a response to our first request
			return exchanges.isEmpty();
		}

		AttrRequests requests = cmParams.getRequests();
		if (requests == null) {
			return true;
		}
		int maxRequests = requests.intValue();
		if (exchanges.size() < maxRequests) {
			return true;
		}
		if (exchanges.size() == maxRequests
				&& (isTermination(msg) || isPause(msg))) {
			// One additional terminate or pause message is allowed
			return true;
		}
		return false;
	}

	/**
	 * Determines whether or not the session is still active.
	 * 
	 * @return {@code true} if it is, {@code false} otherwise
	 */
	private boolean isWorking() {
		assertLocked();

		return procThread != null;
	}

	/**
	 * Blocks until either the message provided becomes immediately sendable or
	 * until the session is terminated.
	 * 
	 * @param msg
	 *            message to evaluate
	 */
	private void blockUntilSendable(final AbstractBody msg) {
		assertLocked();

		while (isWorking() && !isImmediatelySendable(msg)) {
			try {
				notFull.await();
			} catch (InterruptedException intx) {
				LOG.log(Level.FINEST, INTERRUPTED, intx);
			}
		}
	}

	/**
	 * Modifies the specified body message such that it becomes a new BOSH
	 * session creation request.
	 * 
	 * @param rid
	 *            request ID to use
	 * @param orig
	 *            original body to modify
	 * @return modified message which acts as a session creation request
	 */
	private ComposableBody applySessionCreationRequest(final long rid,
			final ComposableBody orig) throws BOSHException {
		assertLocked();

		Builder builder = orig.rebuild();
		builder.setAttribute(Attributes.TO, cfg.getTo());
		builder.setAttribute(Attributes.XML_LANG, cfg.getLang());
		builder.setAttribute(Attributes.VER, AttrVersion.getSupportedVersion()
				.toString());
		builder.setAttribute(Attributes.WAIT, "60");
		builder.setAttribute(Attributes.HOLD, "1");
		builder.setAttribute(Attributes.RID, Long.toString(rid));
		applyRoute(builder);
		applyFrom(builder);
		builder.setAttribute(Attributes.ACK, "1");

		// Make sure the following are NOT present (i.e., during retries)
		builder.setAttribute(Attributes.SID, null);
		return builder.build();
	}

	/**
	 * Applies routing information to the request message who's builder has been
	 * provided.
	 * 
	 * @param builder
	 *            builder instance to add routing information to
	 */
	private void applyRoute(final Builder builder) {
		assertLocked();

		String route = cfg.getRoute();
		if (route != null) {
			builder.setAttribute(Attributes.ROUTE, route);
		}
	}

	/**
	 * Applies the local station ID information to the request message who's
	 * builder has been provided.
	 * 
	 * @param builder
	 *            builder instance to add station ID information to
	 */
	private void applyFrom(final Builder builder) {
		assertLocked();

		String from = cfg.getFrom();
		if (from != null) {
			builder.setAttribute(Attributes.FROM, from);
		}
	}

	/**
	 * Applies existing session data to the outbound request, returning the
	 * modified request.
	 * 
	 * This method assumes the lock is currently held.
	 * 
	 * @param rid
	 *            request ID to use
	 * @param orig
	 *            original/raw request
	 * @return modified request with session information applied
	 */
	private ComposableBody applySessionData(final long rid,
			final ComposableBody orig) throws BOSHException {
		assertLocked();

		Builder builder = orig.rebuild();
		builder.setAttribute(Attributes.SID, cmParams.getSessionID().toString());
		builder.setAttribute(Attributes.RID, Long.toString(rid));
		applyResponseAcknowledgement(builder, rid);
		return builder.build();
	}

	/**
	 * Sets the 'ack' attribute of the request to the value of the highest 'rid'
	 * of a request for which it has already received a response in the case
	 * where it has also received all responses associated with lower 'rid'
	 * values. The only exception is that, after its session creation request,
	 * the client SHOULD NOT include an 'ack' attribute in any request if it has
	 * received responses to all its previous requests.
	 * 
	 * @param builder
	 *            message builder
	 * @param rid
	 *            current request RID
	 */
	private void applyResponseAcknowledgement(final Builder builder,
			final long rid) {
		assertLocked();

		if (responseAck.equals(Long.valueOf(-1L))) {
			// We have not received any responses yet
			return;
		}

		Long prevRID = Long.valueOf(rid - 1L);
		if (responseAck.equals(prevRID)) {
			// Implicit ack
			return;
		}

		builder.setAttribute(Attributes.ACK, responseAck.toString());
	}

	/**
	 * While we are "connected", process received responses.
	 * 
	 * This method is run in the processing thread.
	 */
	private void processMessages() {
		LOG.log(Level.FINEST, "Processing thread starting");
		try {
			HTTPExchange exch;
			do {
				exch = nextExchange();
				if (exch == null) {
					break;
				}

				// Test hook to manipulate what the client sees:
				ExchangeInterceptor interceptor = exchInterceptor.get();
				if (interceptor != null) {
					HTTPExchange newExch = interceptor.interceptExchange(exch);
					if (newExch == null) {
						LOG.log(Level.FINE, "Discarding exchange on request "
								+ "of test hook: RID="
								+ exch.getRequest()
										.getAttribute(Attributes.RID));
						lock.lock();
						try {
							exchanges.remove(exch);
						} finally {
							lock.unlock();
						}
						continue;
					}
					exch = newExch;
				}

				processExchange(exch);
			} while (true);
		} finally {
			LOG.log(Level.FINEST, "Processing thread exiting");
		}

	}

	/**
	 * Get the next message exchange to process, blocking until one becomes
	 * available if nothing is already waiting for processing.
	 * 
	 * @return next available exchange to process, or {@code null} if no
	 *         exchanges are immediately available
	 */
	private HTTPExchange nextExchange() {
		assertUnlocked();

		final Thread thread = Thread.currentThread();
		HTTPExchange exch = null;
		lock.lock();
		try {
			do {
				if (!thread.equals(procThread)) {
					break;
				}
				exch = exchanges.peek();
				if (exch == null) {
					try {
						notEmpty.await();
					} catch (InterruptedException intx) {
						LOG.log(Level.FINEST, INTERRUPTED, intx);
					}
				}
			} while (exch == null);
		} finally {
			lock.unlock();
		}
		return exch;
	}

	/**
	 * Process the next, provided exchange. This is the main processing method
	 * of the receive thread.
	 * 
	 * @param exch
	 *            message exchange to process
	 */
	private void processExchange(final HTTPExchange exch) {
		assertUnlocked();

		HTTPResponse resp;
		AbstractBody body;
		int respCode;
		try {
			resp = exch.getHTTPResponse();
			body = resp.getBody();
			respCode = resp.getHTTPStatus();
		} catch (BOSHException boshx) {
			LOG.log(Level.FINEST, "Could not obtain response", boshx);
			dispose(boshx);
			return;
		} catch (InterruptedException intx) {
			LOG.log(Level.FINEST, INTERRUPTED, intx);
			dispose(intx);
			return;
		}
		fireResponseReceived(body);

		// Process the message with the current session state
		AbstractBody req = exch.getRequest();
		CMSessionParams params;
		List<HTTPExchange> toResend = null;
		lock.lock();
		try {
			// Check for session creation response info, if needed
			if (cmParams == null) {
				cmParams = CMSessionParams.fromSessionInit(req, body);

				// The following call handles the lock. It's not an escape.
				fireConnectionEstablished();
			}
			params = cmParams;

			checkForTerminalBindingConditions(body, respCode);
			if (isTermination(body)) {
				// Explicit termination
				lock.unlock();
				dispose(null);
				return;
			}

			if (isRecoverableBindingCondition(body)) {
				// Retransmit outstanding requests
				if (toResend == null) {
					toResend = new ArrayList<HTTPExchange>(exchanges.size());
				}
				for (HTTPExchange exchange : exchanges) {
					HTTPExchange resendExch = new HTTPExchange(
							exchange.getRequest());
					toResend.add(resendExch);
				}
				for (HTTPExchange exchange : toResend) {
					exchanges.add(exchange);
				}
			} else {
				// Process message as normal
				processRequestAcknowledgements(req, body);
				processResponseAcknowledgementData(req);
				HTTPExchange resendExch = processResponseAcknowledgementReport(body);
				if (resendExch != null && toResend == null) {
					toResend = new ArrayList<HTTPExchange>(1);
					toResend.add(resendExch);
					exchanges.add(resendExch);
				}
			}
		} catch (BOSHException boshx) {
			LOG.log(Level.FINEST, "Could not process response", boshx);
			lock.unlock();
			dispose(boshx);
			return;
		} finally {
			if (lock.isHeldByCurrentThread()) {
				try {
					exchanges.remove(exch);
					if (exchanges.isEmpty()) {
						scheduleEmptyRequest(processPauseRequest(req));
					}
					notFull.signalAll();
				} finally {
					lock.unlock();
				}
			}
		}

		if (toResend != null) {
			for (HTTPExchange resend : toResend) {
				HTTPResponse response = httpSender.send(params,
						resend.getRequest());
				resend.setHTTPResponse(response);
				fireRequestSent(resend.getRequest());
			}
		}
	}

	/**
	 * Clears any scheduled empty requests.
	 */
	private void clearEmptyRequest() {
		assertLocked();

		if (emptyRequestFuture != null) {
			emptyRequestFuture.cancel(false);
			emptyRequestFuture = null;
		}
	}

	/**
	 * Calculates the default empty request delay/interval to use for the active
	 * session.
	 * 
	 * @return delay in milliseconds
	 */
	private long getDefaultEmptyRequestDelay() {
		assertLocked();

		// Figure out how long we should wait before sending an empty request
		AttrPolling polling = cmParams.getPollingInterval();
		long delay;
		if (polling == null) {
			delay = EMPTY_REQUEST_DELAY;
		} else {
			delay = polling.getInMilliseconds();
		}
		return delay;
	}

	/**
	 * Schedule an empty request to be sent if no other requests are sent in a
	 * reasonable amount of time.
	 */
	private void scheduleEmptyRequest(long delay) {
		assertLocked();
		if (delay < 0L) {
			throw (new IllegalArgumentException(
					"Empty request delay must be >= 0 (was: " + delay + ")"));
		}

		clearEmptyRequest();
		if (!isWorking()) {
			return;
		}

		// Schedule the transmission
		if (LOG.isLoggable(Level.FINER)) {
			LOG.finer("Scheduling empty request in " + delay + "ms");
		}
		try {
			emptyRequestFuture = schedExec.schedule(emptyRequestRunnable,
					delay, TimeUnit.MILLISECONDS);
		} catch (RejectedExecutionException rex) {
			LOG.log(Level.FINEST, "Could not schedule empty request", rex);
		}
		drained.signalAll();
	}

	/**
	 * Sends an empty request to maintain session requirements. If a request is
	 * sent within a reasonable time window, the empty request transmission will
	 * be cancelled.
	 */
	private void sendEmptyRequest() {
		assertUnlocked();
		// Send an empty request
		LOG.finest("Sending empty request");
		try {
			send(ComposableBody.builder().build());
		} catch (BOSHException boshx) {
			dispose(boshx);
		}
	}

	/**
	 * Assert that the internal lock is held.
	 */
	private void assertLocked() {
		if (ASSERTIONS) {
			if (!lock.isHeldByCurrentThread()) {
				throw (new AssertionError("Lock is not held by current thread"));
			}
			return;
		}
	}

	/**
	 * Assert that the internal lock is *not* held.
	 */
	private void assertUnlocked() {
		if (ASSERTIONS) {
			if (lock.isHeldByCurrentThread()) {
				throw (new AssertionError("Lock is held by current thread"));
			}
			return;
		}
	}

	/**
	 * Checks to see if the response indicates a terminal binding condition (as
	 * per XEP-0124 section 17). If it does, an exception is thrown.
	 * 
	 * @param body
	 *            response body to evaluate
	 * @param code
	 *            HTTP response code
	 * @throws BOSHException
	 *             if a terminal binding condition is detected
	 */
	private void checkForTerminalBindingConditions(final AbstractBody body,
			final int code) throws BOSHException {
		TerminalBindingCondition cond = getTerminalBindingCondition(code, body);
		if (cond != null) {
			throw (new BOSHException("Terminal binding condition encountered: "
					+ cond.getCondition() + "  (" + cond.getMessage() + ")"));
		}
	}

	/**
	 * Determines whether or not the response indicates a recoverable binding
	 * condition (as per XEP-0124 section 17).
	 * 
	 * @param resp
	 *            response body
	 * @return {@code true} if it does, {@code false} otherwise
	 */
	private static boolean isRecoverableBindingCondition(final AbstractBody resp) {
		return ERROR.equals(resp.getAttribute(Attributes.TYPE));
	}

	/**
	 * Process the request to determine if the empty request delay can be
	 * determined by looking to see if the request is a pause request. If it
	 * can, the request's delay is returned, otherwise the default delay is
	 * returned.
	 * 
	 * @return delay in milliseconds that should elapse prior to an empty
	 *         message being sent
	 */
	private long processPauseRequest(final AbstractBody req) {
		assertLocked();

		if (cmParams != null && cmParams.getMaxPause() != null) {
			try {
				AttrPause pause = AttrPause.createFromString(req
						.getAttribute(Attributes.PAUSE));
				if (pause != null) {
					long delay = pause.getInMilliseconds() - PAUSE_MARGIN;
					if (delay < 0) {
						delay = EMPTY_REQUEST_DELAY;
					}
					return delay;
				}
			} catch (BOSHException boshx) {
				LOG.log(Level.FINEST, "Could not extract", boshx);
			}
		}

		return getDefaultEmptyRequestDelay();
	}

	/**
	 * Check the response for request acknowledgements and take appropriate
	 * action.
	 * 
	 * This method assumes the lock is currently held.
	 * 
	 * @param req
	 *            request
	 * @param resp
	 *            response
	 */
	private void processRequestAcknowledgements(final AbstractBody req,
			final AbstractBody resp) {
		assertLocked();

		if (!cmParams.isAckingRequests()) {
			return;
		}

		// If a report or time attribute is set, we aren't acking anything
		if (resp.getAttribute(Attributes.REPORT) != null) {
			return;
		}

		// Figure out what the highest acked RID is
		String acked = resp.getAttribute(Attributes.ACK);
		Long ackUpTo;
		if (acked == null) {
			// Implicit ack of all prior requests up until RID
			ackUpTo = Long.parseLong(req.getAttribute(Attributes.RID));
		} else {
			ackUpTo = Long.parseLong(acked);
		}

		// Remove the acked requests from the list
		if (LOG.isLoggable(Level.FINEST)) {
			LOG.finest("Removing pending acks up to: " + ackUpTo);
		}
		Iterator<ComposableBody> iter = pendingRequestAcks.iterator();
		while (iter.hasNext()) {
			AbstractBody pending = iter.next();
			Long pendingRID = Long.parseLong(pending
					.getAttribute(Attributes.RID));
			if (pendingRID.compareTo(ackUpTo) <= 0) {
				iter.remove();
			}
		}
	}

	/**
	 * Process the response in order to update the response acknowlegement data.
	 * 
	 * This method assumes the lock is currently held.
	 * 
	 * @param req
	 *            request
	 */
	private void processResponseAcknowledgementData(final AbstractBody req) {
		assertLocked();

		Long rid = Long.parseLong(req.getAttribute(Attributes.RID));
		if (responseAck.equals(Long.valueOf(-1L))) {
			// This is the first request
			responseAck = rid;
		} else {
			pendingResponseAcks.add(rid);
			// Remove up until the first missing response (or end of queue)
			Long whileVal = responseAck;
			while (whileVal.equals(pendingResponseAcks.first())) {
				responseAck = whileVal;
				pendingResponseAcks.remove(whileVal);
				whileVal = Long.valueOf(whileVal.longValue() + 1);
			}
		}
	}

	/**
	 * Process the response in order to check for and respond to any potential
	 * ack reports.
	 * 
	 * This method assumes the lock is currently held.
	 * 
	 * @param resp
	 *            response
	 * @return exchange to transmit if a resend is to be performed, or
	 *         {@code null} if no resend is necessary
	 * @throws BOSHException
	 *             when a a retry is needed but cannot be performed
	 */
	private HTTPExchange processResponseAcknowledgementReport(
			final AbstractBody resp) throws BOSHException {
		assertLocked();

		String reportStr = resp.getAttribute(Attributes.REPORT);
		if (reportStr == null) {
			// No report on this message
			return null;
		}

		Long report = Long.parseLong(reportStr);
		Long time = Long.parseLong(resp.getAttribute(Attributes.TIME));
		if (LOG.isLoggable(Level.FINE)) {
			LOG.fine("Received report of missing request (RID=" + report
					+ ", time=" + time + "ms)");
		}

		// Find the missing request
		Iterator<ComposableBody> iter = pendingRequestAcks.iterator();
		AbstractBody req = null;
		while (iter.hasNext() && req == null) {
			AbstractBody pending = iter.next();
			Long pendingRID = Long.parseLong(pending
					.getAttribute(Attributes.RID));
			if (report.equals(pendingRID)) {
				req = pending;
			}
		}

		if (req == null) {
			throw (new BOSHException("Report of missing message with RID '"
					+ reportStr
					+ "' but local copy of that request was not found"));
		}

		// Resend the missing request
		HTTPExchange exch = new HTTPExchange(req);
		exchanges.add(exch);
		notEmpty.signalAll();
		return exch;
	}

	/**
	 * Notifies all request listeners that the specified request is being sent.
	 * 
	 * @param request
	 *            request being sent
	 */
	private void fireRequestSent(final AbstractBody request) {
		assertUnlocked();

		BOSHMessageEvent event = null;
		for (BOSHClientRequestListener listener : requestListeners) {
			if (event == null) {
				event = BOSHMessageEvent.createRequestSentEvent(this, request);
			}
			try {
				listener.requestSent(event);
			} catch (Exception ex) {
				LOG.log(Level.WARNING, UNHANDLED, ex);
			}
		}
	}

	/**
	 * Notifies all response listeners that the specified response has been
	 * received.
	 * 
	 * @param response
	 *            response received
	 */
	private void fireResponseReceived(final AbstractBody response) {
		assertUnlocked();

		BOSHMessageEvent event = null;
		for (BOSHClientResponseListener listener : responseListeners) {
			if (event == null) {
				event = BOSHMessageEvent.createResponseReceivedEvent(this,
						response);
			}
			try {
				listener.responseReceived(event);
			} catch (Exception ex) {
				LOG.log(Level.WARNING, UNHANDLED, ex);
			}
		}
	}

	/**
	 * Notifies all connection listeners that the session has been successfully
	 * established.
	 */
	private void fireConnectionEstablished() {
		final boolean hadLock = lock.isHeldByCurrentThread();
		if (hadLock) {
			lock.unlock();
		}
		try {
			BOSHClientConnEvent event = null;
			for (BOSHClientConnListener listener : connListeners) {
				if (event == null) {
					event = BOSHClientConnEvent
							.createConnectionEstablishedEvent(this);
				}
				try {
					listener.connectionEvent(event);
				} catch (Exception ex) {
					LOG.log(Level.WARNING, UNHANDLED, ex);
				}
			}
		} finally {
			if (hadLock) {
				lock.lock();
			}
		}
	}

	/**
	 * Notifies all connection listeners that the session has been terminated
	 * normally.
	 */
	private void fireConnectionClosed() {
		assertUnlocked();

		BOSHClientConnEvent event = null;
		for (BOSHClientConnListener listener : connListeners) {
			if (event == null) {
				event = BOSHClientConnEvent.createConnectionClosedEvent(this);
			}
			try {
				listener.connectionEvent(event);
			} catch (Exception ex) {
				LOG.log(Level.WARNING, UNHANDLED, ex);
			}
		}
	}

	/**
	 * Notifies all connection listeners that the session has been terminated
	 * due to the exceptional condition provided.
	 * 
	 * @param cause
	 *            cause of the termination
	 */
	private void fireConnectionClosedOnError(final Throwable cause) {
		assertUnlocked();

		BOSHClientConnEvent event = null;
		for (BOSHClientConnListener listener : connListeners) {
			if (event == null) {
				event = BOSHClientConnEvent.createConnectionClosedOnErrorEvent(
						this, pendingRequestAcks, cause);
			}
			try {
				listener.connectionEvent(event);
			} catch (Exception ex) {
				LOG.log(Level.WARNING, UNHANDLED, ex);
			}
		}
	}

}
